package cn.wangjie.spark.shared

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.util.LongAccumulator
import org.apache.spark.{SparkConf, SparkContext}

/**
 * 基于Spark框架使用Scala语言编程实现词频统计WordCount程序，将符号数据过滤，并统计出现的次数
 * -a. 过滤标点符号数据
 * 使用广播变量
 * -b. 统计出标点符号数据出现次数
 *      使用累加器
 **/
object SparkSharedVariableTest {
	
	def main(args: Array[String]): Unit = {
		// 创建应用程序入口SparkContext实例对象
		val sc: SparkContext = {
			// 1.a 创建SparkConf对象，设置应用的配置信息
			val sparkConf: SparkConf = new SparkConf()
				.setAppName(this.getClass.getSimpleName.stripSuffix("$"))
				.setMaster("local[2]")
			// 1.b 传递SparkConf对象，构建Context实例
			SparkContext.getOrCreate(sparkConf)
		}
		
		// a. 读取文件数据
		val datasRDD: RDD[String] = sc.textFile("datas/filter/datas.input", minPartitions = 2)
		println(s"Count = ${datasRDD.count()}")
		
		// TODO: 字典数据，只要有这些单词就过滤: 特殊字符存储列表List中
		val list: List[String] = List(",", ".","!","#","$","%")
		// TODO: 通过广播变量方式，将list广播到Executor中，被Task使用
		val listBroadcast: Broadcast[List[String]] = sc.broadcast(list)
		
		// TODO: 定义一个累加器，计数非单词字符的个数
		val accumulator: LongAccumulator = sc.longAccumulator("number_accum")
		
		// b. 词频统计
		val resultRDD: RDD[(String, Int)] = datasRDD
            .filter(line => null != line && line.trim.length > 0 )
            .flatMap{line => line.trim.split("\\s+")}
			// 使用filter函数
			.filter{word =>
				// TODO: 获取广播变量的值
				val flag: Boolean = listBroadcast.value.contains(word)
				
				// 如果此单词为字符，累加器加1
				if(flag){
					accumulator.add(1L)
				}
				
				!flag
			}
            .map(word => (word, 1))
            .reduceByKey(_ + _)
		
		// c. 打印结果
		resultRDD.foreach(println)
		println(s"Accum = ${accumulator.value}")
		
		Thread.sleep(1000000)
		
		// 应用程序运行结束，关闭资源
		sc.stop()
	}
	
}
